package io.gitee.tziye.core.msg;

import io.gitee.tziye.RenaiClient;
import io.gitee.tziye.common.RenaiUtil;
import io.vavr.Tuple2;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.Date;
import java.util.List;
import java.util.function.Consumer;

@Slf4j
public class RenaiTopic<T> {

    private final MsgOperator msgOperator;

    public final String topic;
    public final String group;
    public final Class<T> clazz;
    @Setter
    private volatile int count = 100;
    @Setter
    private volatile int interval = 100;
    @Setter
    private volatile int replayCount = 500;

    public RenaiTopic(RenaiClient renaiClient, String topic, String group, Class<T> clazz) {
        this.msgOperator = renaiClient.msg();
        this.group = group;
        this.topic = topic;
        this.clazz = clazz;
    }

    public void produce(T value) {
        msgOperator.produce(topic, "", value);
    }

    public void produce(List<T> values) {
        List<Tuple2<String, T>> msgs = values.stream().map(v -> new Tuple2<>("", v)).toList();
        msgOperator.produce(topic, msgs);
    }

    public void produce(String key, T value) {
        msgOperator.produce(topic, key, value);
    }

    public void produceMsg(List<Tuple2<String, T>> msgs) {
        msgOperator.produce(topic, msgs);
    }

    public List<RenaiMsg> consume() {
        return msgOperator.consume(topic, group, count, new Date());
    }

    public List<RenaiMsg> consume(Date startTime) {
        return msgOperator.consume(topic, group, count, startTime);
    }

    public void ack(RenaiMsg msg) {
        msgOperator.ack(topic, msg.getPt(), group, msg.getId());
    }

    public void consume(Consumer<T> consumer) {
        consume(consumer, null);
    }

    public void consume(Consumer<T> consumer, Date startTime) {
        if (startTime == null) {
            startTime = new Date();
        }
        while (!Thread.interrupted()) {
            RenaiMsg lastMsg = null;
            List<RenaiMsg> msgList = consume(startTime);
            for (RenaiMsg msg : msgList) {
                try {
                    T value = RenaiUtil.parse(msg.getMsg(), clazz);
                    consumer.accept(value);
                    lastMsg = msg;
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    break;
                }
            }
            if (lastMsg != null) {
                ack(lastMsg);
                RenaiUtil.sleepMS(interval);
            } else {
                RenaiUtil.sleepMS(interval * 5);
            }
        }
    }

    public void replay(Consumer<T> consumer, Date startTime, Date endTime) {
        long curId = 0;
        int size = replayCount;
        while (size == replayCount) {
            List<RenaiMsg> renaiMsgs = msgOperator.replay(topic, startTime, endTime, replayCount, curId);
            for (RenaiMsg msg : renaiMsgs) {
                T value = RenaiUtil.parse(msg.getMsg(), clazz);
                consumer.accept(value);
                curId = msg.getId();
            }
            size = renaiMsgs.size();
        }
    }

    public void replay(Consumer<T> consumer, long minId, long maxId) {
        long curId = 0;
        int size = replayCount;
        while (size == replayCount) {
            List<RenaiMsg> renaiMsgs = msgOperator.replay(topic, minId, maxId, replayCount, curId);
            for (RenaiMsg msg : renaiMsgs) {
                T value = RenaiUtil.parse(msg.getMsg(), clazz);
                consumer.accept(value);
                curId = msg.getId();
            }
            size = renaiMsgs.size();
        }
    }

}
